{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Guest House - Hard"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "import findspark\n",
    "import pandas as pd\n",
    "findspark.init()\n",
    "\n",
    "SVR = '192.168.31.31'\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql import Window\n",
    "\n",
    "sc = (SparkSession.builder.appName('app13-3') \n",
    "      .master(f'spark://{SVR}:7077') \n",
    "      .config('spark.sql.warehouse.dir', f'hdfs://{SVR}:9000/user/hive/warehouse') \n",
    "      .config('spark.cores.max', '4') \n",
    "      .config('spark.executor.instances', '1') \n",
    "      .config('spark.executor.cores', '2') \n",
    "      .config('spark.executor.memory', '10g') \n",
    "      .enableHiveSupport().getOrCreate())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "booking = sc.read.table('sqlzoo.booking')\n",
    "guest = sc.read.table('sqlzoo.guest')\n",
    "room = sc.read.table('sqlzoo.room')\n",
    "rate = sc.read.table('sqlzoo.rate')\n",
    "extra = sc.read.table('sqlzoo.extra')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 11.\n",
    "Coincidence. Have two guests with the same surname ever stayed in the hotel on the evening? Show the last name and both first names. Do not include duplicates.\n",
    "\n",
    "```\n",
    "+-----------+------------+-------------+\n",
    "| last_name | first_name | first_name  |\n",
    "+-----------+------------+-------------+\n",
    "| Davies    | Philip     | David T. C. |\n",
    "| Evans     | Graham     | Mr Nigel    |\n",
    "| Howarth   | Mr George  | Sir Gerald  |\n",
    "| Jones     | Susan Elan | Mr Marcus   |\n",
    "| Lewis     | Clive      | Dr Julian   |\n",
    "| McDonnell | John       | Dr Alasdair |\n",
    "+-----------+------------+-------------+\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>last_name</th>\n",
       "      <th>first_name1</th>\n",
       "      <th>first_name</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Davies</td>\n",
       "      <td>Philip</td>\n",
       "      <td>David T. C.</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Evans</td>\n",
       "      <td>Graham</td>\n",
       "      <td>Mr Nigel</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Howarth</td>\n",
       "      <td>Mr George</td>\n",
       "      <td>Sir Gerald</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>Jones</td>\n",
       "      <td>Susan Elan</td>\n",
       "      <td>Mr Marcus</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>Lewis</td>\n",
       "      <td>Clive</td>\n",
       "      <td>Dr Julian</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>McDonnell</td>\n",
       "      <td>John</td>\n",
       "      <td>Dr Alasdair</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   last_name first_name1   first_name\n",
       "0     Davies      Philip  David T. C.\n",
       "1      Evans      Graham     Mr Nigel\n",
       "2    Howarth   Mr George   Sir Gerald\n",
       "3      Jones  Susan Elan    Mr Marcus\n",
       "4      Lewis       Clive    Dr Julian\n",
       "5  McDonnell        John  Dr Alasdair"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "t = (guest.join(booking, on=(guest['id']==booking['guest_id']))\n",
    "     .select('last_name', 'first_name', 'booking_date', 'nights', 'id')\n",
    "     .withColumn('checkout', date_add(col('booking_date'), col('nights')-lit(1))))\n",
    "\n",
    "a = (t.join(t.alias('t1'), 'last_name', how='full')\n",
    "     .toDF('last_name', 'first_name', 'booking_date', 'nights', 'id', 'checkout',\n",
    "           'first_name1', 'booking_date1', 'nights1', 'id1', 'checkout1')\n",
    "     .filter(col('first_name') != col('first_name1')))\n",
    "\n",
    "(a.filter((((col('booking_date') >= col('booking_date1')) & \n",
    "            (col('booking_date') <= col('checkout1'))) |\n",
    "          ((col('booking_date1') >= col('booking_date')) & \n",
    "            (col('booking_date1') <= col('checkout')))) &\n",
    "          (col('id') >= col('id1')))\n",
    " .select('last_name', 'first_name1', 'first_name')\n",
    " .distinct()\n",
    " .orderBy('last_name')\n",
    " .toPandas())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 12.\n",
    "Check out per floor. The first digit of the room number indicates the floor – e.g. room 201 is on the 2nd floor. For each day of the week beginning 2016-11-14 show how many rooms are being vacated that day by floor number. Show all days in the correct order.\n",
    "\n",
    "```\n",
    "+------------+-----+-----+-----+\n",
    "| i          | 1st | 2nd | 3rd |\n",
    "+------------+-----+-----+-----+\n",
    "| 2016-11-14 |   5 |   3 |   4 |\n",
    "| 2016-11-15 |   6 |   4 |   1 |\n",
    "| 2016-11-16 |   2 |   2 |   4 |\n",
    "| 2016-11-17 |   5 |   3 |   6 |\n",
    "| 2016-11-18 |   2 |   3 |   2 |\n",
    "| 2016-11-19 |   5 |   5 |   1 |\n",
    "| 2016-11-20 |   2 |   2 |   2 |\n",
    "+------------+-----+-----+-----+\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>checkout</th>\n",
       "      <th>1st</th>\n",
       "      <th>2nd</th>\n",
       "      <th>3rd</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2016-11-14</td>\n",
       "      <td>5</td>\n",
       "      <td>3</td>\n",
       "      <td>4</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2016-11-15</td>\n",
       "      <td>6</td>\n",
       "      <td>4</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2016-11-16</td>\n",
       "      <td>2</td>\n",
       "      <td>2</td>\n",
       "      <td>4</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2016-11-17</td>\n",
       "      <td>5</td>\n",
       "      <td>3</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>2016-11-18</td>\n",
       "      <td>2</td>\n",
       "      <td>3</td>\n",
       "      <td>2</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>2016-11-19</td>\n",
       "      <td>5</td>\n",
       "      <td>5</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>2016-11-20</td>\n",
       "      <td>2</td>\n",
       "      <td>2</td>\n",
       "      <td>2</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "     checkout  1st  2nd  3rd\n",
       "0  2016-11-14    5    3    4\n",
       "1  2016-11-15    6    4    1\n",
       "2  2016-11-16    2    2    4\n",
       "3  2016-11-17    5    3    6\n",
       "4  2016-11-18    2    3    2\n",
       "5  2016-11-19    5    5    1\n",
       "6  2016-11-20    2    2    2"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(booking\n",
    " .withColumn('checkout', date_add(col('booking_date'), col('nights')))\n",
    " .withColumn('floor', col('room_no').substr(1, 1))\n",
    " .filter(col('checkout').between('2016-11-14', '2016-11-20'))\n",
    " .replace({'1': '1st', '2': '2nd', '3': '3rd', '4': '4th'}, \n",
    "          subset=['floor'])\n",
    " .groupBy('checkout')\n",
    " .pivot('floor')\n",
    " .count()\n",
    " .orderBy('checkout')\n",
    " .toPandas())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 13.\n",
    "Free rooms? List the rooms that are free on the day 25th Nov 2016.\n",
    "\n",
    "```\n",
    "+-----+\n",
    "| id  |\n",
    "+-----+\n",
    "| 207 |\n",
    "| 210 |\n",
    "| 304 |\n",
    "+-----+\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>id</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>207</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>210</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>304</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "    id\n",
       "0  207\n",
       "1  210\n",
       "2  304"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(room.join(\n",
    "    booking\n",
    "    .withColumn('checkout', date_add(col('booking_date'), col('nights')-1))\n",
    "    .filter((col('booking_date') <= '2016-11-25') & \n",
    "            (col('checkout') >= '2016-11-25')),\n",
    "    on=(booking['room_no']==room['id']), how='left_anti')\n",
    " .select('id')\n",
    " .toPandas())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 14.\n",
    "Single room for three nights required. A customer wants a single room for three consecutive nights. Find the first available date in December 2016.\n",
    "\n",
    "```\n",
    "+-----+------------+\n",
    "| id  | MIN(i)     |\n",
    "+-----+------------+\n",
    "| 201 | 2016-12-11 |\n",
    "+-----+------------+\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>room_no</th>\n",
       "      <th>checkout</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>201</td>\n",
       "      <td>2016-12-11</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   room_no    checkout\n",
       "0      201  2016-12-11"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(booking\n",
    " .withColumn('checkout', date_add(col('booking_date'), col('nights')))\n",
    " .join(room.filter(col('room_type')=='single'), \n",
    "       on=(booking['room_no']==room['id']), how='right')\n",
    " .filter(((col('checkout') >= '2016-12-01') & \n",
    "          (col('booking_date') <= '2016-12-31')) | \n",
    "         (col('checkout').isNull()))\n",
    " .withColumn('next_booking', lag('booking_date', 1, '2017-01-01').over(\n",
    "     Window.partitionBy('room_no').orderBy('room_no')))\n",
    " .withColumn('diff', (col('next_booking')-col('checkout')).cast('integer'))\n",
    " .filter(col('diff') >= 3)\n",
    " .select('room_no', 'checkout')\n",
    " .orderBy('checkout')\n",
    " .limit(1)\n",
    " .toPandas())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 15.\n",
    "Gross income by week. Money is collected from guests when they leave. For each Thursday in November and December 2016, show the total amount of money collected from the previous Friday to that day, inclusive.\n",
    "\n",
    "```\n",
    "+------------+---------------+\n",
    "| Thursday   | weekly_income |\n",
    "+------------+---------------+\n",
    "| 2016-11-03 |          0.00 |\n",
    "| 2016-11-10 |      12608.94 |\n",
    "| 2016-11-17 |      13552.56 |\n",
    "| 2016-11-24 |      12929.69 |\n",
    "| 2016-12-01 |      11685.14 |\n",
    "| 2016-12-08 |      13093.79 |\n",
    "| 2016-12-15 |       8975.87 |\n",
    "| 2016-12-22 |       1395.77 |\n",
    "| 2016-12-29 |          0.00 |\n",
    "| 2017-01-05 |          0.00 |\n",
    "+------------+---------------+\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|  Thursday|\n",
      "+----------+\n",
      "|2016-11-03|\n",
      "|2016-11-10|\n",
      "|2016-11-17|\n",
      "|2016-11-24|\n",
      "|2016-12-01|\n",
      "|2016-12-08|\n",
      "|2016-12-15|\n",
      "|2016-12-22|\n",
      "|2016-12-29|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# generate Thursdays\n",
    "import datetime\n",
    "\n",
    "sdate = datetime.date(2016, 11, 1)\n",
    "edate = datetime.date(2016, 12, 31)\n",
    "days_to_thu = (3 - sdate.weekday()) % 7\n",
    "week_diff = ((edate - sdate).days - days_to_thu) // 7\n",
    "\n",
    "thur = sc.createDataFrame([\n",
    "    (sdate + datetime.timedelta(days=days_to_thu + 7 * more_wks), )\n",
    "    for more_wks in range(week_diff + 1)], ['Thursday'])\n",
    "thur.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Thursday</th>\n",
       "      <th>weekly_income</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2016-11-03</td>\n",
       "      <td>0.00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2016-11-10</td>\n",
       "      <td>12608.94</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2016-11-17</td>\n",
       "      <td>13552.56</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2016-11-24</td>\n",
       "      <td>12929.69</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>2016-12-01</td>\n",
       "      <td>11685.14</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>2016-12-08</td>\n",
       "      <td>13093.79</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>2016-12-15</td>\n",
       "      <td>8975.87</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>2016-12-22</td>\n",
       "      <td>1395.77</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>2016-12-29</td>\n",
       "      <td>0.00</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "     Thursday  weekly_income\n",
       "0  2016-11-03           0.00\n",
       "1  2016-11-10       12608.94\n",
       "2  2016-11-17       13552.56\n",
       "3  2016-11-24       12929.69\n",
       "4  2016-12-01       11685.14\n",
       "5  2016-12-08       13093.79\n",
       "6  2016-12-15        8975.87\n",
       "7  2016-12-22        1395.77\n",
       "8  2016-12-29           0.00"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(booking\n",
    " .join(rate, on=(booking['occupants']==rate['occupancy']) & \n",
    "                (booking['room_type_requested']==rate['room_type']), \n",
    "       how='left')\n",
    " .withColumn('income', col('amount') * col('nights'))\n",
    " .select('booking_date', 'nights', 'income')\n",
    " .union(booking\n",
    "        .join(extra, on='booking_id', how='left')\n",
    "        .withColumn('income', col('amount'))\n",
    "        .select('booking_date', 'nights', 'income'))\n",
    " .withColumn('checkout', date_add(col('booking_date'), col('nights')))\n",
    " .withColumn('wkd', dayofweek(col('checkout')))\n",
    " .withColumn('Thursday', when(\n",
    "     col('wkd')<=5, date_add(col('checkout'), lit(5)-col('wkd')))\n",
    "     .otherwise(date_add(col('checkout'), lit(12)-col('wkd'))))\n",
    " .join(thur, on='Thursday', how='right')\n",
    " .groupBy('Thursday')\n",
    " .agg(sum('income').alias('weekly_income'))\n",
    " .fillna(0)\n",
    " .orderBy('Thursday')\n",
    " .toPandas())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
